package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TestQuestionBean;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTradeTrademarkCategoryUserRefundWindow3 {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 设置状态后端
                /*
                env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
                env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
                env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
                env.setStateBackend(new HashMapStateBackend());
                env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
                System.setProperty("HADOOP_USER_NAME", "atguigu");
                 */
        // TODO 3. 从 Kafka dwd_test_test_exam 主题读取考试主题的数据
        String topic = "dwd_test_test_exam";
        String groupId = "dws_trade_trademark_category_user_refund_window3";
        DataStreamSource<String> dbStream = env.addSource(KafkaUtil.getKafkaConsumer(topic, groupId));

        // TODO 4. 转换数据结构
        SingleOutputStreamOperator<JSONObject> mapStream = dbStream.map(JSON::parseObject);
        KeyedStream<JSONObject, String> keyedStream1 = mapStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("question_id")+jsonObject.getString("user_id");
            }
        });
        SingleOutputStreamOperator<TestQuestionBean> javaBeanStream = keyedStream1.process(new ProcessFunction<JSONObject, TestQuestionBean>() {

            ValueState<String> lastVisitState = null;

            public void open(Configuration parameters) throws Exception {
                lastVisitState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last_visit", String.class));
            }

            @Override
            public void processElement(JSONObject jsonObject, Context context, Collector<TestQuestionBean> collector) throws Exception {
                Long ts = jsonObject.getLong("ts");
                String visitDt = DateFormatUtil.toDate(ts * 1000L);
                String lastVisitDt = lastVisitState.value();
                if (lastVisitDt == null || !lastVisitDt.equals(visitDt)) {
                    collector.collect(TestQuestionBean.builder()
                            .examId(jsonObject.getString("id"))
                            .userId(jsonObject.getString("user_id"))
                            .paperId(jsonObject.getString("paper_id"))
                            .questionId(jsonObject.getString("question_id"))
                            .answerCount(1L)
                            .answerCorrectCount("1".equals(jsonObject.getString("is_correct")) ? 1L : 0L)
                            .ts(jsonObject.getLong("ts") * 1000)
                            .build());
                }
            }
        });


        // TODO 6. 设置水位线
        SingleOutputStreamOperator<TestQuestionBean> withWaterMarkStream = javaBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TestQuestionBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner(new SerializableTimestampAssigner<TestQuestionBean>() {
                    @Override
                    public long extractTimestamp(TestQuestionBean tradeTrademarkCategoryUserRefundBean, long l) {
                        return tradeTrademarkCategoryUserRefundBean.getTs();
                    }
                }));



        // TODO 7. 分组
        KeyedStream<TestQuestionBean, String> keyedStream = withWaterMarkStream.keyBy(new KeySelector<TestQuestionBean, String>() {
            @Override
            public String getKey(TestQuestionBean javaBean) throws Exception {
                return javaBean.getQuestionId();
            }
        });

        // TODO 8. 开窗
        WindowedStream<TestQuestionBean, String, TimeWindow> windowStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        // TODO 9. 聚合
        SingleOutputStreamOperator<TestQuestionBean> reduceStream = windowStream.reduce(new ReduceFunction<TestQuestionBean>() {
            @Override
            public TestQuestionBean reduce(TestQuestionBean testBean, TestQuestionBean t1) throws Exception {
                testBean.setAnswerCount(testBean.getAnswerCount()+ t1.getAnswerCount());
                testBean.setAnswerCorrectCount(testBean.getAnswerCorrectCount()+t1.getAnswerCorrectCount());
                testBean.setAnswerCorrectRate(testBean.getAnswerCorrectCount()/(1.0*testBean.getAnswerCount()));
                return testBean;
            }
        }, new WindowFunction<TestQuestionBean, TestQuestionBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<TestQuestionBean> iterable, Collector<TestQuestionBean> collector) throws Exception {
                TestQuestionBean tradeTrademarkCategoryUserRefundBean = iterable.iterator().next();
                tradeTrademarkCategoryUserRefundBean.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                tradeTrademarkCategoryUserRefundBean.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                tradeTrademarkCategoryUserRefundBean.setTs(System.currentTimeMillis());
                collector.collect(tradeTrademarkCategoryUserRefundBean);
            }
        });

        reduceStream.print("reduce>>>");


        env.execute(groupId);


    }
}
